package com.htdata.iiot;

import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.alibaba.fastjson.JSONObject;
import com.htdata.iiot.api.pojo.DeviceNumber;
import com.htdata.iiot.api.pojo.StreamingData;
import com.htdata.iiot.api.pojo.StreamingMetricData;

/**
 * 状态数据压力测试
 * 
 */
public class Main12 {
	private static Logger logger = LogManager.getLogger(Main12.class);
	private static int times = 60;

	public static void main(String[] args) throws ParseException {
		Cmd.parserCmd(args);
		times = Cmd.times;
		long pageSize = 10000;
		if (Cmd.pageSize > 0) {
			pageSize = Cmd.pageSize;
		}

		List<StreamingData> sList = new ArrayList<StreamingData>();
		SendDataToKafka sendDataToKafka = new SendDataToKafka();
		for (int i = 0; i < pageSize; i++) {
			int deviceId = i;
			int tenant_id = i % 100;

			StreamingData streamingData = new StreamingData();
			streamingData.setDid(deviceId);
			streamingData.setTid(tenant_id);
			List<StreamingMetricData> mList = new ArrayList<StreamingMetricData>();
			String name = "status";
			long mId = deviceId;
			StreamingMetricData metricData = new StreamingMetricData();
			metricData.setK(name);
			metricData.setDn(new DeviceNumber(tenant_id, deviceId, mId).toHexString());

			mList.add(metricData);
			streamingData.setData(mList);
			sList.add(streamingData);
		}

		logger.info("****************************");
		logger.info(sList.size());
		logger.info("****************************");
		long startTime = System.currentTimeMillis() - 1;
		long st = System.currentTimeMillis();
		long i = 0;
		long mi = 0;
		long avg = sList.size();
		if(times>0){
			avg = sList.size() / (times * 2);
		}
		if (avg == 0) {
			avg = sList.size();
		}

		while (true) {

			// 3.生成数据
			for (StreamingData sData : sList) {
				List<StreamingMetricData> data = sData.getData();
				for (StreamingMetricData mData : data) {
					mData.setV(((System.currentTimeMillis()%2)+1)*1000);
					mData.setTs(System.currentTimeMillis());
					mi++;
				}

				sData.setTs(System.currentTimeMillis());
				sendDataToKafka.send(sData.getDid(),buildResult(sData));
				i++;
				if (i % avg == 0) {
					// 线程休息
					try {
						logger.info("中场休息" + "i:" + i);
						Thread.sleep(400);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}

			long endTime = System.currentTimeMillis();
			long useTime = endTime - startTime;
			logger.info(st + "写入 " + i + " 条数据,累计测点数：" + mi + "，平均速率 " + (i * 1000 / useTime) + "，测点速率 "
					+ (mi * 1000 / useTime));
			if (endTime < (st + times * 1000)) {
				try {
					logger.info("本场结束");
					Thread.sleep(st + times * 1000 - endTime);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			} else {
				logger.info("本场超时");
			}

		}

	}

	private static String buildResult(StreamingData sData) {
		return JSONObject.toJSONString(sData).toString();
	}

}